package com.cellulam.framework.mq.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.cellulam.framework.mq.ActionResult;
import lombok.extern.slf4j.Slf4j;

import java.util.Properties;

@Slf4j
public abstract class MqMessageListener
        extends BaseAbstractMessageListener<Consumer> {

    protected MqMessageListener() {
        this(MqConsumeMessageMode.CLUSTERING);
    }

    protected MqMessageListener(MqConsumeMessageMode messageMode) {
        super(messageMode);
    }

    @Override
    protected final Consumer createConsumer(Properties properties) {
        return ONSFactory.createConsumer(properties);
    }

    @Override
    public final void subscribe(String groupId, String topic, String subExpression) {
        log.info("订阅, groupId: {}, topic: {}, subExpression: {}",
                groupId, topic, subExpression);

        super.getConsumer().subscribe(topic, subExpression, ((message, context) -> {
            ActionResult result = super.consumeMessage(message);
            return this.convertToOrderAction(result);
        }));
    }

    private Action convertToOrderAction(ActionResult result) {
        switch (result) {
            case SUCCESS:
                return Action.CommitMessage;
            default:
                return Action.ReconsumeLater;
        }
    }
}
